﻿//namespace NTLib.Pipe.Network.Containers
//{
//    using NTLib.Pipe.Containers;
//    using System;
//    using System.Collections.Generic;
//    using System.Linq;
//    using System.Text;
//    using System.Threading.Tasks;
//    using System.Threading;
//    using NTLib.Pipe.Network.Descriptors;
//    using NTLib.Pipe.Exceptions;
//    using NTLib.Pipe.Network.Connectors;
//    using NTLib.Pipe.Network.Formats;
//    using NTLib.Pipe.Network.Exceptions;
//    using NTLib.Core.Threading;
//    using NTLib.Core.ScopeLock;
//    using System.Runtime.CompilerServices;
//    using NTLib.Core.Component;

//    // <summary>
//    // Pipe used to transfert data througth network
//    // </summary>
//    public class NetworkPipeSource<T> : BasePipeSourcePoolItem<T, NetworkTargetDescription>
//    {
//        #region Fields

//        private readonly AsyncLock _locker;
//        private readonly ReaderWriterLocker _resourceAccess;
//        private readonly AsyncLock _sendLocker;

//        private readonly Queue<IPackage> _toSend;
//        private readonly Dictionary<Guid, Tuple<Type, TaskCompletionSource<object>>> _answers;
//        private readonly Queue<IPackage> _read;
//        private Queue<TaskCompletionSource<object>> _readWaiter;

//        private bool _connectorConcurrent;
//        private IConnector _connector;
//        private INetworkPackageFormatter _formater;
//        private CancellationTokenSource _actionCancelSource;

//        #endregion

//        #region CTOR

//        /// <summary>
//        /// Initializes a new instance of the <see cref="NetworkPipeSource{T}"/> class.
//        /// </summary>
//        public NetworkPipeSource()
//        {
//            this._resourceAccess = new ReaderWriterLocker();
//            this._locker = new AsyncLock();
//            this._toSend = new Queue<IPackage>();
//            this._sendLocker = new AsyncLock();
//            this._read = new Queue<IPackage>();
//            this._answers = new Dictionary<Guid, Tuple<Type, TaskCompletionSource<object>>>();
//            this._readWaiter = new Queue<TaskCompletionSource<object>>();
//        }

//        #endregion

//        #region Properties

//        /// <summary>
//        /// Gets a value indicating whether this instance can pop item.
//        /// </summary>
//        /// <value>
//        /// <c>true</c> if this instance can pop item; otherwise, <c>false</c>.
//        /// </value>
//        public override bool CanPopItem
//        {
//            get
//            {
//                CheckInitialized();

//                using (this._resourceAccess.ScopeLock(true))
//                {
//                    if (this._connector != null)
//                        return this._connector.CanPopItem && (this.PipeType & PipeType.FromTarget) == PipeType.FromTarget; ;
//                    return false;
//                }
//            }
//        }

//        /// <summary>
//        /// Gets the value indicating that the current pipe allow the user to push items through it.
//        /// </summary>
//        /// <value>
//        /// <c>true</c> if this instance can push item; otherwise, <c>false</c>.
//        /// </value>
//        public override bool CanPushItem
//        {
//            get
//            {
//                CheckInitialized();
//                using (this._resourceAccess.ScopeLock(true))
//                {
//                    if (this._connector != null)
//                        return this._connector.CanPushItem && (this.PipeType & PipeType.OneWay) == PipeType.OneWay;
//                    return false;
//                }
//            }
//        }

//        #endregion

//        #region Methods

//        /// <summary>
//        /// Initializes the current pipe asynchronously.
//        /// </summary>
//        /// <param name="token"></param>
//        /// <returns>
//        /// Task used to make this initialization asynchronously.
//        /// </returns>
//        /// <exception cref="NTLib.Pipe.Exceptions.InvalidTargetTypeException"></exception>
//        /// <exception cref="NTLib.Pipe.Exceptions.PipeAlreadyInitializedException"></exception>
//        protected override async Task OnInitializeAsync(CancellationToken token = default(CancellationToken))
//        {
//            await base.OnInitializeAsync(token);

//            if (this._connector != null && this._formater != null)
//                throw new PipeAlreadyInitializedException();

//            if (this.TargetTyped.Connector == null)
//                throw new MissingConnectorException();

//            if (this.TargetTyped.Formater == null)
//                throw new MissingFormaterException();

//            try
//            {
//                // Get Connector
//                this._connector = ConnectorManager.Instance.Create(this.TargetTyped.Connector.Id);

//                // Get Formator
//                this._formater = FormatManager.Instance.Create(this.TargetTyped.Formater.Id);

//                if (!this._connector.IsInitialized)
//                    await this._connector.InitializeAsync(this.TargetTyped.IsHost);

//                this._connectorConcurrent = Core.Reflection.TypeExtensions.HasAttr<ConcurrentComponentAttribute>(this._connector, true);

//                this._actionCancelSource = new CancellationTokenSource();

//                if (this.TargetTyped.IsHost) {
//                    var a = this._connector.ConnectAsync(this.TargetTyped.Address, this._actionCancelSource.Token).ConfigureAwait(false);
//                }

//                Status = PipeStatus.Ready;
//            }
//            catch
//            {
//                this._connector = null;
//                this._formater = null;
//                throw;
//            }

//            if (this.CanPopItem)
//                ReadLoop();
//        }


//        /// <summary>
//        /// Called when the update pop waiter is called.
//        /// This methods is used to dequeue item from the queue to the waiter.
//        /// This calls is Thread Safe.
//        /// </summary>
//        /// <param name="token"></param>
//        /// <returns></returns>
//        protected override async Task<T> OnDequeueItemAsync(CancellationToken token)
//        {
//            CheckInitialized();

//            TaskCompletionSource<object> _order = null;
//            IPackage pkg = null;
//            using (await this._locker.LockAsync(token))
//            {
//                if (this._read.Count > 0 && this._readWaiter.Count == 0)
//                    pkg = this._read.Dequeue();
//                else
//                {
//                    _order = new TaskCompletionSource<object>();
//                    this._readWaiter.Enqueue(_order);
//                }
//            }

//            if (pkg == null && _order != null)
//            {
//                await _order.Task;
//                pkg = (IPackage)_order.Task.Result;
//            }

//            using (this._resourceAccess.ScopeLock(true))
//            {
//                return this._formater.FromPackage<T>(pkg);
//            }
//        }

//        protected override async Task OnPushItemAsync<U>(T item, TaskCompletionSource<object> waiter, CancellationToken token)
//        {
//            CheckInitialized();
//            using (await this._locker.LockAsync(token))
//            {
//                IPackage pkg = this._formater.ToPackage<T>(item);
//                this._toSend.Enqueue(pkg);
//                this._answers.Add(pkg.Id, new Tuple<Type, TaskCompletionSource<object>>(typeof(U), waiter));
//            }
//            SendLoop();
//        }

//        protected override async Task OnSend(T item, CancellationToken token)
//        {
//            CheckInitialized();
//            using (await this._locker.LockAsync(token))
//            {
//                this._toSend.Enqueue(this._formater.ToPackage<T>(item));
//            }
//            SendLoop();
//        }

//        /// <summary>
//        /// Checks if the instance is initialized.
//        /// </summary>
//        [MethodImpl(MethodImplOptions.AggressiveInlining)]
//        private void CheckInitialized()
//        {
//            if (this._connector == null)
//                throw new PipeNotInitializedException();
//        }

//        /// <summary>
//        /// Checks if the instance is initialized.
//        /// </summary>
//        [MethodImpl(MethodImplOptions.AggressiveInlining)]
//        private bool CheckConnectionEstablish()
//        {
//            if (this._connector == null)
//            {
//                return false;
//            }

//            return this._connector.Connected == ConnectionStatus.Connected;
//        }

//        protected override void ManagedDispose()
//        {
//            using (this._resourceAccess.ScopeLock())
//            {
//                if (this._actionCancelSource != null)
//                {
//                    this._actionCancelSource.Cancel();
//                    this._actionCancelSource = null;
//                }

//                if (this._connector != null)
//                {
//                    var tmp = this._connector;
//                    this._connector = null;
//                    tmp.Dispose();
//                }

//                if (this._formater != null)
//                {
//                    var tmp = this._formater;
//                    this._formater = null;
//                    tmp.Dispose();
//                }

//                this._read.Clear();
//                this._toSend.Clear();
//                this._answers.Clear();

//                Status = PipeStatus.Closed;
//            }
//        }

//        private async void ReadLoop()
//        {
//            try
//            {
//                IConnector connector;
//                INetworkPackageFormatter formatter;
//                CancellationTokenSource source;
//                do
//                {
//                    using (this._resourceAccess.ScopeLock(true))
//                    {
//                        connector = this._connector;
//                        formatter = this._formater;
//                        source = this._actionCancelSource;
//                    }

//                    if (source != null && connector != null && formatter != null)
//                    {
//                        source.Token.ThrowIfCancellationRequested();

//                        if (connector.Connected != ConnectionStatus.Connected)
//                            await connector.ConnectAsync(this.TargetTyped.Address, source.Token);

//                        try
//                        {
//                            var packageRead = await connector.ReadAsync(formatter.ReaderDescriptor, source.Token);
//                            if (packageRead != null)
//                            {
//                                using (await this._locker.LockAsync(source.Token))
//                                {
//                                    source.Token.ThrowIfCancellationRequested();

//                                    Tuple<Type, TaskCompletionSource<object>> waiter;
//                                    if (this._answers.TryGetValue(packageRead.Id, out waiter))
//                                    {
//                                        waiter.Item2.TrySetResult(formatter.FromPackage(packageRead, waiter.Item1));
//                                        this._answers.Remove(packageRead.Id);
//                                    }
//                                    else
//                                    {
//                                        if (this._readWaiter.Count == 0)
//                                        {
//                                            this._read.Enqueue(packageRead);
//                                        }
//                                        else
//                                        {
//                                            var readWaiter = this._readWaiter.Dequeue();
//                                            readWaiter.TrySetResult(packageRead);
//                                            continue;
//                                        }
//                                    }
//                                }

//                                source.Token.ThrowIfCancellationRequested();
//                                base.RaiseDataReceived();
//                            }
//                        }
//                        catch (PipeException)
//                        {

//                        }
//                    }
//                } while (connector != null && source != null);
//            }
//            catch (OperationCanceledException)
//            {
//            }
//        }

//        private async void SendLoop()
//        {
//            if (this._connectorConcurrent)
//            {
//                await SendItemsThreadUnsafe();
//            }
//            else
//            {
//                using (await _sendLocker.LockAsync())
//                {
//                    await SendItemsThreadUnsafe();
//                }
//            }
//        }

//        private async Task SendItemsThreadUnsafe()
//        {
//            IPackage pkgToSend = null;
//            IConnector connector;
//            CancellationTokenSource source = null;

//            do
//            {
//                using (this._resourceAccess.ScopeLock(true))
//                {
//                    connector = this._connector;
//                    source = this._actionCancelSource;
//                }

//                if (source != null && connector != null)
//                {
//                    using (await this._locker.LockAsync())
//                    {
//                        if (this._toSend.Count > 0)
//                            pkgToSend = this._toSend.Peek();
//                        else
//                            pkgToSend = null;

//                    }

//                    if (pkgToSend != null)
//                    {
//                        source.Token.ThrowIfCancellationRequested();
//                        if (connector.Connected != ConnectionStatus.Connected)
//                            await connector.ConnectAsync(this.TargetTyped.Address, source.Token);

//                        source.Token.ThrowIfCancellationRequested();
//                        await connector.SendAsync(pkgToSend, source.Token);

//                        this._toSend.Dequeue();
//                        pkgToSend.Dispose();
//                    }

//                    source.Token.ThrowIfCancellationRequested();
//                }
//            } while (pkgToSend != null);
//        }

//        /// <summary>
//        /// Raises the data received.
//        /// </summary>
//        protected override void RaiseDataReceived()
//        {
//            CancellationTokenSource source = null;
//            using (this._resourceAccess.ScopeLock(true))
//            {
//                source = this._actionCancelSource;
//            }
//            var lockerTask = this._locker.LockAsync(source.Token);
//            lockerTask.Wait(source.Token);
//            using (lockerTask.Result)
//            {
//                if (this._readWaiter.Count > 0)
//                {
//                    var readWaiter = this._readWaiter.Dequeue();
//                    readWaiter.SetResult(42);
//                    return;
//                }
//            }
//            base.RaiseDataReceived();
//        }

//        #endregion
//    }
//}
